package Flink;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.List;

/**
 * 先map，再设置水印
 * @Author: leiwenkang02
 * @Date: 2020/3/10 10:00 下午
 */
@Slf4j
public class WatermarkDemo02 {
    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> text = env.socketTextStream("localhost", 9000, "\n");

        text.map((MapFunction<String, WordWithTime>) value -> {
            String[] array = value.split("\\s");
            WordWithTime wordWithTime = new WordWithTime(array[0], Long.parseLong(array[1]) * 1000);
            System.out.println("wordWithTime: " + wordWithTime);
            return wordWithTime;
        }).assignTimestampsAndWatermarks((new AssignerWithPeriodicWatermarks<WordWithTime>() {
            private static final long serialVersionUID = 8252616297345284790L;
            Long currentMaxTimestamp = 0L;
            Long maxOutOfOrderness = 3000L;//最大允许的乱序时间
            Watermark watermark = null;
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                watermark = new Watermark(currentMaxTimestamp - maxOutOfOrderness);
                return watermark;
            }

            @Override
            public long extractTimestamp(WordWithTime wordWithTime, long previousElementTimestamp) {
                long timestamp = wordWithTime.getTimestamp();
                currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
                System.out.println("timestamp : " + wordWithTime.getTimestamp() + "|" + format.format(wordWithTime.getTimestamp())
                        + " currentMaxTimestamp : " + currentMaxTimestamp + "|" + format.format(currentMaxTimestamp)
                        + "watermark : " + watermark);
                return timestamp;
            }
        })).keyBy(WordWithTime::getWord)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<WordWithTime, WordWithTime, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<WordWithTime> iterable, Collector<WordWithTime> collector) throws Exception {
                        Iterator<WordWithTime> iterator = iterable.iterator();
                        List<WordWithTime> wordWithTimes = Lists.newArrayList();
                        while (iterator.hasNext()) {
                            WordWithTime wordWithTime = iterator.next();
                            wordWithTimes.add(wordWithTime);
                        }
                        System.out.println("ProcessWindowFunction, wordWithTime: " + wordWithTimes);
                        collector.collect(wordWithTimes.get(wordWithTimes.size() - 1));
                    }
                }).addSink(new RichSinkFunction<WordWithTime>() {
            @Override
            public void invoke(WordWithTime wordWithTime, Context context) throws Exception {
                System.out.println("RichSinkFunction, wordWithTime: " + wordWithTime);
            }
        });
        System.out.println("windowCounts.print()");

        env.execute("leiwenkang");
    }

    // Data type for words with count
    public static class WordWithTime {

        public String word;
        public long timestamp;

        public WordWithTime() {
        }

        public WordWithTime(String word, long timestamp) {
            this.word = word;
            this.timestamp = timestamp;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public long getTimestamp() {
            return timestamp;
        }

        public void setTimestamp(long timestamp) {
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return word + " : " + timestamp;
        }
    }
}
